import java.sql.PreparedStatement
import java.text.SimpleDateFormat
import java.util.{Date, Properties}

import com.alibaba.fastjson.JSON
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.jdbc.{JdbcConnectionOptions, JdbcExecutionOptions, JdbcSink, JdbcStatementBuilder}
import org.apache.flink.streaming.api.{CheckpointingMode, TimeCharacteristic}
import org.apache.flink.streaming.api.environment.CheckpointConfig
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011


case class maidian
(distinct_id :String,lib1 :String,lib_method :String,lib_version :String,propertie_lib :String,propertie_version :String,
 network_type :String,manufacturer :String,model :String,screen_width :String,screen_height :String,os :String,os_version :String,
 timezone_offset :String,app_id :String,sreen_id :String,latest_scene :String,url_path :String,event_duration :String,appName :String,
 appId :String,is_first_day :String,anonymous_id :String,type1 :String,event :String,page_id :String,login_id :String,
time :String,dt :String,
 channel_name :String,idmap_reason :String,channel_link_id :String,local_time :String,monetary :String,
 site_language :String,site_id :String,category_id :String,commodity_spu :String,commodity_sku :String,is_new :String,track_signup_original_id :String,
share_content :String,second_category :String,room_name :String,room_id :String,room_category :String,return_price_total :String,
 rebate_amount :String,player_name :String,player_id :String,order_type :String,order_should_pay :String,oa_level :String,
 member_level :String,live_duration :String,is_own_goods :String,is_follow_anchor :String,inviter_id :String,group_type :String,group_number :String
 ,
 group_member :String,goods_price :String,goods_position :String,first_category :String,enter_source :String,duration :String,distributor_level :String,
 distributor_fans :String,distibute_content :String,cost :String,barrage_content :String,anchor_type :String,StoreCouponAmount :String,
 PlatformCouponAmount :String,MixStoreCouponAmount :String,mp_event_key :String,sf_total_msgs_success :String,sf_total_msgs :String,
 mp_reply_appid :String,sf_keyword :String,sf_msg_group_id :String,mp_card_id :String,mp_tail_content :String,mp_option_list :String,
 mp_head_content :String,mp_msg_group_type :String,sf_keyword_rule_id :String,mp_bizmsgmenu_id :String,mp_menu_id :String,mp_url :String,
 mp_description :String,mp_title :String,mp_thumb_media_id :String,mp_media_id :String,mp_pic_url :String,mp_msg_id :String,mp_content :String,
 mp_click_menu_type :String,mp_receive_msg_type :String,app_type_info :String,bot_name :String,sf_convert_id :String,sf_strategy_unit_id :String,
 short_url_target :String,short_url_key :String,element_path :String,app_version :String,browser :String,browser_version :String,carrier :String,
 city :String,country :String,device_id :String,element_class_name :String,element_content :String,element_id :String,element_name :String,element_position :String,
 element_selector :String,element_target_url :String,element_type :String,latest_referrer :String,latest_search_keyword :String,
 latest_traffic_source_type :String,latest_utm_campaign :String,latest_utm_content :String,latest_utm_medium :String,latest_utm_source :String,
 latest_utm_term :String,referrer :String,referrer_host :String,resume_from_background :String,scene :String,is_first_time :String
)
/*case class maidian1(sf_audience_id :String,sf_channel_category :String,sf_channel_id :String,sf_channel_service_name :String,sf_enter_plan_time :String,sf_link_url :String,
                    sf_msg_content :String,sf_msg_id :String,sf_msg_title :String,sf_plan_id :String,sf_plan_strategy_id :String,share_depth :String,share_distinct_id :String,
                    share_url_path :String,title :String,url :String,url_host:String,url_query :String,utm_campaign :String,utm_content :String,utm_matching_type :String,
                    utm_medium :String,utm_source :String,utm_term :String,viewport_height :String,viewport_position :String,viewport_width :String,wifi :String,
                    App_crashed_reason :String,account :String,activity_id :String,activity_name :String,app_name :String,application_id :String,apply_reason :String,
                    banner_belong_area :String,banner_id :String,banner_name :String,banner_rank :String,banner_type :String,cancel_reason :String,cash_back_amount :String,
                    cash_back_name :String,commodity_amount :String,commodity_colour :String,commodity_detail_souce :String,commodity_discount_amount :String,
                    commodity_id :String,commodity_name :String,commodity_quantity :String,commodity_score :String,commodity_size :String,commodity_tag :String,
                    contact_type :String,convert_id :String,delivery_method :String,discount_amount :String,discount_id :String,discount_name :String,discount_price :String,
                    discount_type :String,entrance :String,express_name :String,fail_reason :String,first_commodity :String,friend_num :String,get_type :String,group_id :String,
                    head_id :String,head_name :String,if_use_integral :String,integral_discount_amount :String,invisitor_id :String,is_cash_back :String,is_default_confirm :String,
                    is_login :String,is_quick_login :String,is_success :String,is_use_discount :String,is_vip :String,key_word :String,key_word_classify :String,key_word_type :String,
                    login_method :String,mkt_location :String,mkt_name :String,mkt_page :String,mkt_type :String,monetary_unit :String,number_of_integral :String,offlineStoreID :String,
                    offlineStoreName :String,order_actual_amount:String,order_amount :String,order_id :String,original_price :String,page_redirect :String,page_title :String,
                    page_type :String,payment_method :String,platform_type :String,position_number :String,present_price :String,pushID :String,push_content :String,push_content_id :String,
                    push_type :String,quantity_unit :String,receiver_address :String,receiver_area :String,receiver_city :String,receiver_name :String,
                    receiver_province :String,register_method :String,result_number :String,scanActivityName :String,scanType :String,second_commodity :String,
                    segmentation_name :String,service_type :String,share_type :String,store_id :String,store_name :String,supplier_id :String,supplier_name :String,
                    total_price_of_commodity :String,transportation_costs :String,viewpoint_height :String,vip_level :String)*/

//重写JdbcStatementBuilder
class CHSinkBuilder extends JdbcStatementBuilder[maidian]{
  override def accept(t: PreparedStatement, u: maidian): Unit = {
    t.setString(1, u.distinct_id)
    t.setString(2, u.lib1)
    t.setString(3, u.lib_method)
    t.setString(4, u.lib_version)
    t.setString(5, u.propertie_lib)
    t.setString(6, u.propertie_version)
    t.setString(7, u.network_type)
    t.setString(8, u.manufacturer)
    t.setString(9, u.model)
    t.setString(10, u.screen_width)
    t.setString(11, u.screen_height)
    t.setString(12, u.os)
    t.setString(13, u.os_version)
    t.setString(14, u.timezone_offset)
    t.setString(15, u.app_id)
    t.setString(16, u.sreen_id)
    t.setString(17, u.latest_scene)
    t.setString(18, u.url_path)
    t.setString(19, u.event_duration)
    t.setString(20, u.appName)
    t.setString(21, u.appId)
    t.setString(22, u.is_first_day)
    t.setString(23, u.anonymous_id)
    t.setString(24, u.type1)
    t.setString(25, u.event)
    t.setString(26, u.page_id)
    t.setString(27, u.login_id)
    t.setString(28, u.time)
    t.setString(29, u.dt)
    t.setString(30,u.channel_name)
    t.setString(31,u.idmap_reason)
    t.setString(32,u.channel_link_id)
    t.setString(33,u.local_time)
    t.setString(34,u.monetary)
    t.setString(35,u.site_language)
    t.setString(36,u.site_id)
    t.setString(37,u.category_id)
    t.setString(38,u.commodity_spu)
    t.setString(39,u.commodity_sku)
    t.setString(40,u.is_new)
    t.setString(41,u.track_signup_original_id)
    t.setString(42,u.share_content)
    t.setString(43,u.second_category)
    t.setString(44,u.room_name)
    t.setString(45,u.room_id)
    t.setString(46,u.room_category)
    t.setString(47,u.return_price_total)
    t.setString(48,u.rebate_amount)
    t.setString(49,u.player_name)
    t.setString(50,u.player_id)
    t.setString(51,u.order_type)
    t.setString(52,u.order_should_pay)
    t.setString(53,u.oa_level)
    t.setString(54,u.member_level)
    t.setString(55,u.live_duration)
    t.setString(56,u.is_own_goods)
    t.setString(57,u.is_follow_anchor)
    t.setString(58,u.inviter_id)
    t.setString(59,u.group_type)
    t.setString(60,u.group_number)
    t.setString(61,u.group_member)
    t.setString(62,u.goods_price)
    t.setString(63,u.goods_position)
    t.setString(64,u.first_category)
    t.setString(65,u.enter_source)
    t.setString(66,u.duration)
    t.setString(67,u.distributor_level)
    t.setString(68,u.distributor_fans)
    t.setString(69,u.distibute_content)
    t.setString(70,u.cost)
    t.setString(71,u.barrage_content)
    t.setString(72,u.anchor_type)
    t.setString(73,u.StoreCouponAmount)
    t.setString(74,u.PlatformCouponAmount)
    t.setString(75,u.MixStoreCouponAmount)
    t.setString(76,u.mp_event_key)
    t.setString(77,u.sf_total_msgs_success)
    t.setString(78,u.sf_total_msgs)
    t.setString(79,u.mp_reply_appid)
    t.setString(80,u.sf_keyword)
    t.setString(81,u.sf_msg_group_id)
    t.setString(82,u.mp_card_id)
    t.setString(83,u.mp_tail_content)
    t.setString(84,u.mp_option_list)
    t.setString(85,u.mp_head_content)
    t.setString(86,u.mp_msg_group_type)
    t.setString(87,u.sf_keyword_rule_id)
    t.setString(88,u.mp_bizmsgmenu_id)
    t.setString(89,u.mp_menu_id)
    t.setString(90,u.mp_url)
    t.setString(91,u.mp_description)
    t.setString(92,u.mp_title)
    t.setString(93,u.mp_thumb_media_id)
    t.setString(94,u.mp_media_id)
    t.setString(95,u.mp_pic_url)
    t.setString(96,u.mp_msg_id)
    t.setString(97,u.mp_content)
    t.setString(98,u.mp_click_menu_type)
    t.setString(99,u.mp_receive_msg_type)
    t.setString(100,u.app_type_info)
    t.setString(101,u.bot_name)
    t.setString(102,u.sf_convert_id)
    t.setString(103,u.sf_strategy_unit_id)
    t.setString(104,u.short_url_target)
    t.setString(105,u.short_url_key)
    t.setString(106,u.element_path)
    t.setString(107,u.app_version)
    t.setString(108,u.browser)
    t.setString(109,u.browser_version)
    t.setString(110,u.carrier)
    t.setString(111,u.city)
    t.setString(112,u.country)
    t.setString(113,u.device_id)
    t.setString(114,u.element_class_name)
    t.setString(115,u.element_content)
    t.setString(116,u.element_id)
    t.setString(117,u.element_name)
    t.setString(118,u.element_position)
    t.setString(119,u.element_selector)
    t.setString(120,u.element_target_url)
    t.setString(121,u.element_type)
    t.setString(122,u.latest_referrer)
    t.setString(123,u.latest_search_keyword)
    t.setString(124,u.latest_traffic_source_type)
    t.setString(125,u.latest_utm_campaign)
    t.setString(126,u.latest_utm_content)
    t.setString(127,u.latest_utm_medium)
    t.setString(128,u.latest_utm_source)
    t.setString(129,u.latest_utm_term)
    t.setString(130,u.referrer)
    t.setString(131,u.referrer_host)
    t.setString(132,u.resume_from_background)
    t.setString(133,u.scene)
    t.setString(134,u.is_first_time)
  }
}

object FlinkSinkKafka {

  def main(args: Array[String]): Unit = {
    //获取流处理
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //隐式转换
    import org.apache.flink.api.scala._
    //checkpoint配置
    env.enableCheckpointing(100);
    env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
    env.getCheckpointConfig.setMinPauseBetweenCheckpoints(500);
    env.getCheckpointConfig.setCheckpointTimeout(60000);
    env.getCheckpointConfig.setMaxConcurrentCheckpoints(1);
    env.getCheckpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);;

    //设置并行度
    env.setParallelism(1)

    //设置EventTime
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    //连接kafka
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "183.3.221.132:9092")  //183.3.221.132:9092
    properties.setProperty("group.id", "test")
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

    //获取kafka生产者消息
    val stream = env.addSource(new FlinkKafkaConsumer011[String]("maidian-data", new SimpleStringSchema(), properties))
    //stream.print()

    val skinSql =
      """
        |INSERT INTO events(
        |       distinct_id,lib1,lib_method,lib_version,propertie_lib,propertie_version,
        |       network_type,manufacturer,model,screen_width,screen_height,os,os_version,
        |       timezone_offset,app_id,sreen_id,latest_scene,url_path,event_duration,appName,
        |       appId,is_first_day,anonymous_id,type1,event,page_id,login_id,time,dt,
        |       channel_name,idmap_reason,channel_link_id,local_time,monetary,
        |        site_language,site_id,category_id,commodity_spu,commodity_sku,is_new,track_signup_original_id,
        |        share_content,second_category,room_name,room_id,room_category,return_price_total,
        |        rebate_amount,player_name,player_id,order_type,order_should_pay,oa_level,
        |        member_level,live_duration,is_own_goods,is_follow_anchor,inviter_id,group_type,group_number,
        |        group_member,goods_price,goods_position,first_category,enter_source,duration,distributor_level,
        |        distributor_fans,distibute_content,cost,barrage_content,anchor_type,StoreCouponAmount,
        |        PlatformCouponAmount,MixStoreCouponAmount,mp_event_key,sf_total_msgs_success,sf_total_msgs,
        |        mp_reply_appid,sf_keyword,sf_msg_group_id,mp_card_id,mp_tail_content,mp_option_list,
        |        mp_head_content,mp_msg_group_type,sf_keyword_rule_id,mp_bizmsgmenu_id,mp_menu_id,mp_url,
        |        mp_description,mp_title,mp_thumb_media_id,mp_media_id,mp_pic_url,mp_msg_id,mp_content,
        |        mp_click_menu_type,mp_receive_msg_type,app_type_info,bot_name,sf_convert_id,sf_strategy_unit_id,
        |        short_url_target,short_url_key,element_path,app_version,browser,browser_version,carrier,
        |        city,country,device_id,element_class_name,element_content,element_id,element_name,element_position,
        |        element_selector,element_target_url,element_type,latest_referrer,latest_search_keyword,
        |        latest_traffic_source_type,latest_utm_campaign,latest_utm_content,latest_utm_medium,latest_utm_source,
        |        latest_utm_term,referrer,referrer_host,resume_from_background,scene,is_first_time
        |) VALUES(
        |?,?,?,?,?,
        |?,?,?,?,?,
        |?,?,?,?,?,
        |?,?,?,?,?,
        |?,?,?,?,?,
        |?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,
        |?,?,?,?,?
        |)
        |""".stripMargin


    val DateStream = stream.map(line => {
      val str = line.replace("$", "")
      val distinct_id = JSON.parseObject(str).getString("distinct_id")
      val lib = JSON.parseObject(str).getString("lib")
      val lib1 = JSON.parseObject(lib).getString("lib")
      val lib_method = JSON.parseObject(lib).getString("lib_method")
      val lib_version = JSON.parseObject(lib).getString("lib_version")
      val properties = JSON.parseObject(str).getString("properties")
      val propertie_lib = JSON.parseObject(properties).getString("lib")
      val propertie_version = JSON.parseObject(properties).getString("lib_version")
      val network_type = JSON.parseObject(properties).getString("network_type")
      val manufacturer = JSON.parseObject(properties).getString("manufacturer")
      val model = JSON.parseObject(properties).getString("model")
      val screen_width = JSON.parseObject(properties).getString("screen_width")
      val screen_height = JSON.parseObject(properties).getString("screen_height")
      val os = JSON.parseObject(properties).getString("os")
      val os_version = JSON.parseObject(properties).getString("os_version")
      val timezone_offset = JSON.parseObject(properties).getString("timezone_offset")
      val app_id = JSON.parseObject(properties).getString("app_id")
      val sreen_id = JSON.parseObject(properties).getString("sreen_id")
      val latest_scene = JSON.parseObject(properties).getString("latest_scene")
      val url_path = JSON.parseObject(properties).getString("url_path")
      val event_duration = JSON.parseObject(properties).getString("event_duration")
      val appName = JSON.parseObject(properties).getString("appName")
      val appId = JSON.parseObject(properties).getString("appId")
      val is_first_day = JSON.parseObject(properties).getString("is_first_day")
      val anonymous_id = JSON.parseObject(str).getString("anonymous_id")
      val type1 = JSON.parseObject(str).getString("type")
      val event = JSON.parseObject(str).getString("event")
      val page_id = JSON.parseObject(properties).getString("page_id")
      val login_id = JSON.parseObject(str).getString("login_id")
      val time = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date)
      val dt = new SimpleDateFormat("yyyyMMdd").format(new Date)

      val channel_name = JSON.parseObject(properties).getString("channel_name")
      val idmap_reason = JSON.parseObject(properties).getString("idmap_reason")
      val channel_link_id = JSON.parseObject(properties).getString("channel_link_id")
      val local_time = JSON.parseObject(properties).getString("local_time")
      val monetary = JSON.parseObject(properties).getString("monetary")
      val site_language = JSON.parseObject(properties).getString("site_language")
      val site_id = JSON.parseObject(properties).getString("site_id")
      val category_id = JSON.parseObject(properties).getString("category_id")
      val commodity_spu = JSON.parseObject(properties).getString("commodity_spu")
      val commodity_sku = JSON.parseObject(properties).getString("commodity_sku")
      val is_new = JSON.parseObject(properties).getString("is_new")
      val track_signup_original_id = JSON.parseObject(properties).getString("track_signup_original_id")
      val share_content = JSON.parseObject(properties).getString("share_content")
      val second_category = JSON.parseObject(properties).getString("second_category")
      val room_name = JSON.parseObject(properties).getString("room_name")
      val room_id = JSON.parseObject(properties).getString("room_id")
      val room_category = JSON.parseObject(properties).getString("room_category")
      val return_price_total = JSON.parseObject(properties).getString("return_price_total")
      val rebate_amount = JSON.parseObject(properties).getString("rebate_amount")
      val player_name = JSON.parseObject(properties).getString("player_name")
      val player_id = JSON.parseObject(properties).getString("player_id")
      val order_type = JSON.parseObject(properties).getString("order_type")
      val order_should_pay = JSON.parseObject(properties).getString("order_should_pay")
      val oa_level = JSON.parseObject(properties).getString("oa_level")
      val member_level = JSON.parseObject(properties).getString("member_level")
      val live_duration = JSON.parseObject(properties).getString("live_duration")
      val is_own_goods = JSON.parseObject(properties).getString("is_own_goods")
      val is_follow_anchor = JSON.parseObject(properties).getString("is_follow_anchor")

      val inviter_id = JSON.parseObject(properties).getString("inviter_id")
      val group_type = JSON.parseObject(properties).getString("group_type")
      val group_number = JSON.parseObject(properties).getString("group_number")
      val group_member = JSON.parseObject(properties).getString("group_member")
      val goods_price = JSON.parseObject(properties).getString("goods_price")
      val goods_position = JSON.parseObject(properties).getString("goods_position")
      val first_category = JSON.parseObject(properties).getString("first_category")
      val enter_source = JSON.parseObject(properties).getString("enter_source")
      val duration = JSON.parseObject(properties).getString("duration")
      val distributor_level = JSON.parseObject(properties).getString("distributor_level")
      val distributor_fans = JSON.parseObject(properties).getString("distributor_fans")
      val distibute_content = JSON.parseObject(properties).getString("distibute_content")
      val cost = JSON.parseObject(properties).getString("cost")
      val barrage_content = JSON.parseObject(properties).getString("barrage_content")
      val anchor_type = JSON.parseObject(properties).getString("anchor_type")
      val StoreCouponAmount = JSON.parseObject(properties).getString("StoreCouponAmount")
      val PlatformCouponAmount = JSON.parseObject(properties).getString("PlatformCouponAmount")
      val MixStoreCouponAmount = JSON.parseObject(properties).getString("MixStoreCouponAmount")
      val mp_event_key = JSON.parseObject(properties).getString("mp_event_key")
      val sf_total_msgs_success = JSON.parseObject(properties).getString("sf_total_msgs_success")
      val sf_total_msgs = JSON.parseObject(properties).getString("sf_total_msgs")
      val mp_reply_appid = JSON.parseObject(properties).getString("mp_reply_appid")
      val sf_keyword = JSON.parseObject(properties).getString("sf_keyword")
      val sf_msg_group_id = JSON.parseObject(properties).getString("sf_msg_group_id")
      val mp_card_id = JSON.parseObject(properties).getString("mp_card_id")
      val mp_tail_content = JSON.parseObject(properties).getString("mp_tail_content")
      val mp_option_list = JSON.parseObject(properties).getString("mp_option_list")
      val mp_head_content = JSON.parseObject(properties).getString("mp_head_content")
      val mp_msg_group_type = JSON.parseObject(properties).getString("mp_msg_group_type")
      val sf_keyword_rule_id = JSON.parseObject(properties).getString("sf_keyword_rule_id")
      val mp_bizmsgmenu_id = JSON.parseObject(properties).getString("mp_bizmsgmenu_id")
      val mp_menu_id = JSON.parseObject(properties).getString("mp_menu_id")
      val mp_url = JSON.parseObject(properties).getString("mp_url")
      val mp_description = JSON.parseObject(properties).getString("mp_description")
      val mp_title = JSON.parseObject(properties).getString("mp_title")
      val mp_thumb_media_id = JSON.parseObject(properties).getString("mp_thumb_media_id")
      val mp_media_id = JSON.parseObject(properties).getString("mp_media_id")
      val mp_pic_url = JSON.parseObject(properties).getString("mp_pic_url")
      val mp_msg_id = JSON.parseObject(properties).getString("mp_msg_id")
      val mp_content = JSON.parseObject(properties).getString("mp_content")
      val mp_click_menu_type = JSON.parseObject(properties).getString("mp_click_menu_type")
      val mp_receive_msg_type = JSON.parseObject(properties).getString("mp_receive_msg_type")
      val app_type_info = JSON.parseObject(properties).getString("app_type_info")
      val bot_name = JSON.parseObject(properties).getString("bot_name")
      val sf_convert_id = JSON.parseObject(properties).getString("sf_convert_id")

      val sf_strategy_unit_id = JSON.parseObject(properties).getString("sf_strategy_unit_id")
      val short_url_target = JSON.parseObject(properties).getString("short_url_target")
      val short_url_key = JSON.parseObject(properties).getString("short_url_key")
      val element_path = JSON.parseObject(properties).getString("element_path")
      val app_version = JSON.parseObject(properties).getString("app_version")
      val browser = JSON.parseObject(properties).getString("browser")
      val browser_version = JSON.parseObject(properties).getString("browser_version")
      val carrier = JSON.parseObject(properties).getString("carrier")
      val city = JSON.parseObject(properties).getString("city")
      val country = JSON.parseObject(properties).getString("country")
      val device_id = JSON.parseObject(properties).getString("device_id")
      val element_class_name = JSON.parseObject(properties).getString("element_class_name")
      val element_content = JSON.parseObject(properties).getString("element_content")
      val element_id = JSON.parseObject(properties).getString("element_id")
      val element_name = JSON.parseObject(properties).getString("element_name")
      val element_position = JSON.parseObject(properties).getString("element_position")
      val element_selector = JSON.parseObject(properties).getString("element_selector")
      val element_target_url = JSON.parseObject(properties).getString("element_target_url")
      val element_type = JSON.parseObject(properties).getString("element_type")
      val ip = JSON.parseObject(properties).getString("ip")
      val is_first_time = JSON.parseObject(properties).getString("is_first_time")
      val latest_referrer = JSON.parseObject(properties).getString("latest_referrer")
      val latest_search_keyword = JSON.parseObject(properties).getString("latest_search_keyword")
      val latest_traffic_source_type = JSON.parseObject(properties).getString("latest_traffic_source_type")
      val latest_utm_campaign = JSON.parseObject(properties).getString("latest_utm_campaign")
      val latest_utm_content = JSON.parseObject(properties).getString("latest_utm_content")
      val latest_utm_medium = JSON.parseObject(properties).getString("latest_utm_medium")
      val latest_utm_source = JSON.parseObject(properties).getString("latest_utm_source")
      val latest_utm_term = JSON.parseObject(properties).getString("latest_utm_term")

      val province = JSON.parseObject(properties).getString("province")
      val referrer = JSON.parseObject(properties).getString("referrer")
      val referrer_host = JSON.parseObject(properties).getString("referrer_host")
      val resume_from_background = JSON.parseObject(properties).getString("resume_from_background")
      val scene = JSON.parseObject(properties).getString("scene")
      val screen_name = JSON.parseObject(properties).getString("screen_name")
      val sf_audience_id = JSON.parseObject(properties).getString("sf_audience_id")
      val sf_channel_category = JSON.parseObject(properties).getString("sf_channel_category")
      val sf_channel_id = JSON.parseObject(properties).getString("sf_channel_id")
      val sf_channel_service_name = JSON.parseObject(properties).getString("sf_channel_service_name")
      val sf_enter_plan_time = JSON.parseObject(properties).getString("sf_enter_plan_time")
      val sf_link_url = JSON.parseObject(properties).getString("sf_link_url")
      val sf_msg_content = JSON.parseObject(properties).getString("sf_msg_content")
      val sf_msg_id = JSON.parseObject(properties).getString("sf_msg_id")
      val sf_msg_title = JSON.parseObject(properties).getString("sf_msg_title")
      val sf_plan_id = JSON.parseObject(properties).getString("sf_plan_id")
      val sf_plan_strategy_id = JSON.parseObject(properties).getString("sf_plan_strategy_id")
      val share_depth = JSON.parseObject(properties).getString("share_depth")
      val share_distinct_id = JSON.parseObject(properties).getString("share_distinct_id")
      val share_url_path = JSON.parseObject(properties).getString("share_url_path")
      val title = JSON.parseObject(properties).getString("title")
      val url = JSON.parseObject(properties).getString("url")
      val url_host = JSON.parseObject(properties).getString("url_host")
      val url_query = JSON.parseObject(properties).getString("url_query")
      val utm_campaign = JSON.parseObject(properties).getString("utm_campaign")
      val utm_content = JSON.parseObject(properties).getString("utm_content")
      val utm_matching_type = JSON.parseObject(properties).getString("utm_matching_type")
      val utm_medium = JSON.parseObject(properties).getString("utm_medium")
      val utm_source = JSON.parseObject(properties).getString("utm_source")
      val utm_term = JSON.parseObject(properties).getString("utm_term")
      val viewport_height = JSON.parseObject(properties).getString("viewport_height")
      val viewport_position = JSON.parseObject(properties).getString("viewport_position")
      val viewport_width = JSON.parseObject(properties).getString("viewport_width")
      val wifi = JSON.parseObject(properties).getString("wifi")
      val App_crashed_reason = JSON.parseObject(properties).getString("App_crashed_reason")
      val account = JSON.parseObject(properties).getString("account")
      val activity_id = JSON.parseObject(properties).getString("activity_id")
      val activity_name = JSON.parseObject(properties).getString("activity_name")
      val app_name = JSON.parseObject(properties).getString("app_name")
      val application_id = JSON.parseObject(properties).getString("application_id")
      val apply_reason = JSON.parseObject(properties).getString("apply_reason")
      val banner_belong_area = JSON.parseObject(properties).getString("banner_belong_area")
      val banner_id = JSON.parseObject(properties).getString("banner_id")
      val banner_name = JSON.parseObject(properties).getString("banner_name")
      val banner_rank = JSON.parseObject(properties).getString("banner_rank")
      val banner_type = JSON.parseObject(properties).getString("banner_type")
      val cancel_reason = JSON.parseObject(properties).getString("cancel_reason")
      val cash_back_amount = JSON.parseObject(properties).getString("cash_back_amount")
      val cash_back_name = JSON.parseObject(properties).getString("cash_back_name")
      val commodity_amount = JSON.parseObject(properties).getString("commodity_amount")
      val commodity_colour = JSON.parseObject(properties).getString("commodity_colour")
      val commodity_detail_souce = JSON.parseObject(properties).getString("commodity_detail_souce")
      val commodity_discount_amount = JSON.parseObject(properties).getString("commodity_discount_amount")
      val commodity_id = JSON.parseObject(properties).getString("commodity_id")
      val commodity_name = JSON.parseObject(properties).getString("commodity_name")
      val commodity_quantity = JSON.parseObject(properties).getString("commodity_quantity")
      val commodity_score = JSON.parseObject(properties).getString("commodity_score")
      val commodity_size = JSON.parseObject(properties).getString("commodity_size")
      val commodity_tag = JSON.parseObject(properties).getString("commodity_tag")
      val contact_type = JSON.parseObject(properties).getString("contact_type")
      val convert_id = JSON.parseObject(properties).getString("convert_id")
      val delivery_method = JSON.parseObject(properties).getString("delivery_method")
      val discount_amount = JSON.parseObject(properties).getString("discount_amount")
      val discount_id = JSON.parseObject(properties).getString("discount_id")
      val discount_name = JSON.parseObject(properties).getString("discount_name")

      val discount_price = JSON.parseObject(properties).getString("discount_price")
      val discount_type = JSON.parseObject(properties).getString("discount_type")
      val entrance = JSON.parseObject(properties).getString("entrance")
      val express_name = JSON.parseObject(properties).getString("express_name")
      val fail_reason = JSON.parseObject(properties).getString("fail_reason")
      val first_commodity = JSON.parseObject(properties).getString("first_commodity")
      val friend_num = JSON.parseObject(properties).getString("friend_num")
      val get_type = JSON.parseObject(properties).getString("get_type")
      val group_id = JSON.parseObject(properties).getString("group_id")
      val head_id = JSON.parseObject(properties).getString("head_id")
      val head_name = JSON.parseObject(properties).getString("head_name")
      val if_use_integral = JSON.parseObject(properties).getString("if_use_integral")
      val integral_discount_amount = JSON.parseObject(properties).getString("integral_discount_amount")
      val invisitor_id = JSON.parseObject(properties).getString("invisitor_id")
      val is_cash_back = JSON.parseObject(properties).getString("is_cash_back")
      val is_default_confirm = JSON.parseObject(properties).getString("is_default_confirm")
      val is_login = JSON.parseObject(properties).getString("is_login")
      val is_quick_login = JSON.parseObject(properties).getString("is_quick_login")
      val is_success = JSON.parseObject(properties).getString("is_success")
      val is_use_discount = JSON.parseObject(properties).getString("is_use_discount")
      val is_vip = JSON.parseObject(properties).getString("is_vip")
      val key_word = JSON.parseObject(properties).getString("key_word")
      val key_word_classify = JSON.parseObject(properties).getString("key_word_classify")
      val key_word_type = JSON.parseObject(properties).getString("key_word_type")
      val login_method = JSON.parseObject(properties).getString("login_method")
      val mkt_location = JSON.parseObject(properties).getString("mkt_location")
      val mkt_name = JSON.parseObject(properties).getString("mkt_name")
      val mkt_page = JSON.parseObject(properties).getString("mkt_page")
      val mkt_type = JSON.parseObject(properties).getString("mkt_type")
      val monetary_unit = JSON.parseObject(properties).getString("monetary_unit")
      val number_of_integral = JSON.parseObject(properties).getString("number_of_integral")
      val offlineStoreID = JSON.parseObject(properties).getString("offlineStoreID")
      val offlineStoreName = JSON.parseObject(properties).getString("offlineStoreName")
      val order_actual_amount = JSON.parseObject(properties).getString("order_actual_amount")
      val order_amount = JSON.parseObject(properties).getString("order_amount")
      val order_id = JSON.parseObject(properties).getString("order_id")
      val original_price = JSON.parseObject(properties).getString("original_price")
      val page_redirect = JSON.parseObject(properties).getString("page_redirect")

      val page_title = JSON.parseObject(properties).getString("page_title")
      val page_type = JSON.parseObject(properties).getString("page_type")
      val payment_method = JSON.parseObject(properties).getString("payment_method")
      val platform_type = JSON.parseObject(properties).getString("platform_type")
      val position_number = JSON.parseObject(properties).getString("position_number")
      val present_price = JSON.parseObject(properties).getString("present_price")
      val pushID = JSON.parseObject(properties).getString("pushID")
      val push_content = JSON.parseObject(properties).getString("push_content")
      val push_content_id = JSON.parseObject(properties).getString("push_content_id")
      val push_type = JSON.parseObject(properties).getString("push_type")
      val quantity_unit = JSON.parseObject(properties).getString("quantity_unit")
      val receiver_address = JSON.parseObject(properties).getString("receiver_address")
      val receiver_area = JSON.parseObject(properties).getString("receiver_area")
      val receiver_city = JSON.parseObject(properties).getString("receiver_city")
      val receiver_name = JSON.parseObject(properties).getString("receiver_name")
      val receiver_province = JSON.parseObject(properties).getString("receiver_province")
      val register_method = JSON.parseObject(properties).getString("register_method")
      val result_number = JSON.parseObject(properties).getString("result_number")
      val scanActivityName = JSON.parseObject(properties).getString("scanActivityName")
      val scanType = JSON.parseObject(properties).getString("scanType")
      val second_commodity = JSON.parseObject(properties).getString("second_commodity")
      val segmentation_name = JSON.parseObject(properties).getString("segmentation_name")
      val service_type = JSON.parseObject(properties).getString("service_type")
      val share_type = JSON.parseObject(properties).getString("share_type")
      val store_id = JSON.parseObject(properties).getString("store_id")
      val store_name = JSON.parseObject(properties).getString("store_name")
      val supplier_id = JSON.parseObject(properties).getString("supplier_id")
      val supplier_name = JSON.parseObject(properties).getString("supplier_name")
      val total_price_of_commodity = JSON.parseObject(properties).getString("total_price_of_commodity")
      val transportation_costs = JSON.parseObject(properties).getString("transportation_costs")
      val viewpoint_height = JSON.parseObject(properties).getString("viewpoint_height")
      val vip_level = JSON.parseObject(properties).getString("vip_level")




      maidian(distinct_id,lib1,lib_method,lib_version,propertie_lib,propertie_version,
        network_type,manufacturer,model,screen_width,screen_height,os,os_version,
        timezone_offset,app_id,sreen_id,latest_scene,url_path,event_duration,appName,
        appId,is_first_day,anonymous_id,type1,event,page_id,login_id,time,dt,
        channel_name,idmap_reason,channel_link_id,local_time,monetary,
        site_language,site_id,category_id,commodity_spu,commodity_sku,is_new,track_signup_original_id,
        share_content,second_category,room_name,room_id,room_category,return_price_total,
        rebate_amount,player_name,player_id,order_type,order_should_pay,oa_level,
        member_level,live_duration,is_own_goods,is_follow_anchor,inviter_id,group_type,group_number,
        group_member,goods_price,goods_position,first_category,enter_source,duration,distributor_level,
        distributor_fans,distibute_content,cost,barrage_content,anchor_type,StoreCouponAmount,
        PlatformCouponAmount,MixStoreCouponAmount,mp_event_key,sf_total_msgs_success,sf_total_msgs,
        mp_reply_appid,sf_keyword,sf_msg_group_id,mp_card_id,mp_tail_content,mp_option_list,
        mp_head_content,mp_msg_group_type,sf_keyword_rule_id,mp_bizmsgmenu_id,mp_menu_id,mp_url,
        mp_description,mp_title,mp_thumb_media_id,mp_media_id,mp_pic_url,mp_msg_id,mp_content,
        mp_click_menu_type,mp_receive_msg_type,app_type_info,bot_name,sf_convert_id,sf_strategy_unit_id,
        short_url_target,short_url_key,element_path,app_version,browser,browser_version,carrier,
        city,country,device_id,element_class_name,element_content,element_id,element_name,element_position,
        element_selector,element_target_url,element_type,latest_referrer,latest_search_keyword,
        latest_traffic_source_type,latest_utm_campaign,latest_utm_content,latest_utm_medium,latest_utm_source,
        latest_utm_term,referrer,referrer_host,resume_from_background,scene,is_first_time
      )
     /* maidian1(sf_audience_id,sf_channel_category,sf_channel_id,sf_channel_service_name,sf_enter_plan_time,sf_link_url,
        sf_msg_content,sf_msg_id,sf_msg_title,sf_plan_id,sf_plan_strategy_id,share_depth,share_distinct_id,
        share_url_path,title,url,url_host ,url_query,utm_campaign,utm_content,utm_matching_type,
        utm_medium,utm_source,utm_term,viewport_height,viewport_position,viewport_width,wifi,
        App_crashed_reason,account,activity_id,activity_name,app_name,application_id,apply_reason,
        banner_belong_area,banner_id,banner_name,banner_rank,banner_type,cancel_reason,cash_back_amount,
        cash_back_name,commodity_amount,commodity_colour,commodity_detail_souce,commodity_discount_amount,
        commodity_id,commodity_name,commodity_quantity,commodity_score,commodity_size,commodity_tag,
        contact_type,convert_id,delivery_method,discount_amount,discount_id,discount_name,discount_price,
        discount_type,entrance,express_name,fail_reason,first_commodity,friend_num,get_type,group_id,
        head_id,head_name,if_use_integral,integral_discount_amount,invisitor_id,is_cash_back,is_default_confirm,
        is_login,is_quick_login,is_success,is_use_discount,is_vip,key_word,key_word_classify,key_word_type,
        login_method,mkt_location,mkt_name,mkt_page,mkt_type,monetary_unit,number_of_integral,offlineStoreID,
        offlineStoreName,order_actual_amount ,order_amount,order_id,original_price,page_redirect,page_title,
        page_type,payment_method,platform_type,position_number,present_price,pushID,push_content,push_content_id,
        push_type,quantity_unit,receiver_address,receiver_area,receiver_city,receiver_name,
        receiver_province,register_method,result_number,scanActivityName,scanType,second_commodity,
        segmentation_name,service_type,share_type,store_id,store_name,supplier_id,supplier_name,
        total_price_of_commodity,transportation_costs,viewpoint_height,vip_level)*/
    })
    DateStream.print()



    DateStream.addSink(
      JdbcSink.sink(
        skinSql,
        new CHSinkBuilder,
        new JdbcExecutionOptions.Builder()
          .withBatchSize(1)
          .build(),
        new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
          .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
          .withUrl("jdbc:clickhouse://106.14.31.66:8123/xiaoxiangai")
          .withUsername("default")
          .withPassword("Ssyt@118")
          .build()
      )
    )

    env.execute()
  }
}
